package cn.seaboot.admin.activemq;

import cn.seaboot.commons.file.IOUtils;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.ActiveMQSession;

import javax.jms.*;
import java.nio.charset.StandardCharsets;


/**
 * 测试消息队列
 *
 * @author Mr.css
 * @version 2024-10-12 11:09
 */
public class TcpReceiver {

    private static final String USERNAME = "admin";
    private static final String PASSWORD = "admin";
    private static final String BROKER_URL = "tcp://localhost:61616";
    private static final String TOPIC_NAME = "mqtt";

    public static void main(String[] args) throws Exception {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(USERNAME, PASSWORD, BROKER_URL);
        // 连接
        Connection connection = factory.createConnection();

        // 设置客户端 ID,向 MQ 服务器注册自己的名称
        connection.setClientID("client_a");

        // 会话 - 自动确认
        Session session = connection.createSession(false, ActiveMQSession.CLIENT_ACKNOWLEDGE);
//        Session session = connection.createSession(false, ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE);
        Topic topic = session.createTopic(TOPIC_NAME);

        // 创建一个 topic 订阅者对象
        TopicSubscriber topicSubscriber = session.createDurableSubscriber(topic, "playWithUs");

        connection.start();

        while (true){
            // receive() 函数如果不带参数，则表示持续侦听消息
            Message message = topicSubscriber.receive();

            System.out.println(message.getClass());

            if (message instanceof TextMessage) {
                TextMessage msg = (TextMessage) message;
                msg.acknowledge();
                System.out.println("Received message: " + msg.getText());
            } else if(message instanceof BytesMessage) {
                BytesMessage msg = (BytesMessage) message;

                StringBuilder sb = new StringBuilder();
                byte[] buffer = new byte[IOUtils.DEFAULT_BUFFER_SIZE];
                int len = buffer.length;
                while ((len = msg.readBytes(buffer, len)) > 0) {
                    sb.append(new String(buffer, 0, len, StandardCharsets.UTF_8));
                }
                System.out.println(sb.toString());
                msg.acknowledge();
            } else {
                break;
            }
        }

        session.close();
        connection.close();
    }
}
